/**
* <p>Title: MQConsumer.java</p>
* <p>Description: </p>
* <p>Company: www.huoyunren.com</p>
* @author duanxiangyun
* @date 2016年11月11日
* @version 1.0
*/
package com.hxbd.test.utils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MQConsumer {
	private StopEventPushConsumer rocketPushSpout;
	/**
	 * 构造函数
	 * @param mqUrl mq nameserver地址
	 * @param consumerGroup 消费者组,自己定义，不能与已配的组重名，不然会消费该组消息，影响业务
	 * @param topicStr topic
	 * @param tagStr tag
	 * @param imei 以imei作为key,传null会查询所有消息
	 */
	public MQConsumer (String mqUrl, String consumerGroup, String topicStr,String tagStr,String imei) {
		Map<String, String> conf = new HashMap<String, String>();
		conf.put(StopEventPushConsumer.NAMESERVER, mqUrl);
		conf.put(StopEventPushConsumer.CONSUMERGROUP, consumerGroup);
		conf.put(StopEventPushConsumer.TOPIC, topicStr);
		conf.put(StopEventPushConsumer.SUBTAG, tagStr);
		conf.put(StopEventPushConsumer.KEY, imei);
		StopEventPushConsumer rocketPushSpout1 = new StopEventPushConsumer(conf);
		rocketPushSpout = rocketPushSpout1;
	}
	/**
	 * 获得MQ数据	 
	 * @param minsize 最小得到数据条数
	 * @return list
	 */
	public List<?> getMsg(int minsize){		
		System.out.println("Press enter/return to quit\n");
		int index = 0;
		int count = 0;
		while (true) {
			List<?> l = rocketPushSpout.getMsgList();
			int tmpcount = l.size();
			if (tmpcount >= minsize) {
				rocketPushSpout.close();
				return l;
			}
			else {
				if (index == tmpcount) {
					count++;
				}
				else {
					count = 0;
				}
				if (count == 3) {
					System.out.println("消费超过30秒数据没有变化，断开");
					return l;
				}
				index = tmpcount;
				try {
					Thread.sleep(10000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
		
	}
}
